package com.shujia.spark.streaming

import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("wc")
    val sc = new SparkContext(conf)

    //创建sparkStraming上下文对象,  需要指定创建RDD的间隔时间    batch时间   可以理解为多少秒计算一次
    val ssc = new StreamingContext(sc, Durations.seconds(5))


    //设置checkpoint的路径
    ssc.checkpoint("spark/data/wc")


    //连接实时数据源，创建DStream,
    //相当于socket的client
    //可以在linux中启动一个服务端    ，  nc -lk 8888
    val linesDS: DStream[String] = ssc.socketTextStream("node1", 8888)

    /**
      * Dstream的代码逻辑，每个5秒会被计算一次，每次计算都是独立的
      *
      */
    val kvDS: DStream[(String, Int)] = linesDS
      .flatMap(_.split(","))
      .map((_, 1))

    /**
      *
      * 有状态计算
      * reduceByKey只能进行局部统计，每一个批次统计一次
      *
      * updateStateByKey  更新每一个key的状态（单词的数量）
      *
      */

    /**
      * @param seq   当前batch一个key的所有value
      * @param count 之前所有batch的计算结果，  之前的状态存放在checkpoint目录下
      *
      *
      * updataValue:每一个batch每一个key调用一次
      */

    def updateValue(seq: Seq[Int], count: Option[Int]) = {

      //获取之前统计的结果
      val last = count.getOrElse(0)

      //当前batch单词的数量
      val curr = seq.sum

      //更新之后的状态（单词的数量）
      Option(curr + last)
    }


    //累加统计单词数量
    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updateValue)


    //相当于一个action算子
    countDS.print()

    //启动sparkstreaming程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
